// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <pollux/expression/cast_expr.h>

#include <fmt/format.h>
#include <stdexcept>

#include <pollux/common/base/exceptions.h>
#include <pollux/core/core_type_system.h>
#include <pollux/expression/peeled_encoding.h>
#include <pollux/expression/presto_cast_hooks.h>
#include <pollux/expression/scoped_var_setter.h>
#include <pollux/functions/lib/rows_translation_util.h>
#include <pollux/type/type.h>
#include <pollux/type/tz/time_zone_map.h>
#include <pollux/vector/complex_vector.h>
#include <pollux/vector/function_vector.h>
#include <pollux/vector/selectivity_vector.h>

namespace kumo::pollux::exec {
    std::string_view
    detail::extractDigits(const char *s, size_t start, size_t size) {
        size_t pos = start;
        for (; pos < size; ++pos) {
            if (!std::isdigit(s[pos])) {
                break;
            }
        }
        return std::string_view(s + start, pos - start);
    }

    Status detail::parseDecimalComponents(
        const char *s,
        size_t size,
        detail::DecimalComponents &out) {
        if (size == 0) {
            return Status::UserError("Input is empty.");
        }

        size_t pos = 0;

        // Sign of the number.
        if (s[pos] == '-') {
            out.sign = -1;
            ++pos;
        } else if (s[pos] == '+') {
            out.sign = 1;
            ++pos;
        }

        // Extract the whole digits.
        out.wholeDigits = detail::extractDigits(s, pos, size);
        pos += out.wholeDigits.size();
        if (pos == size) {
            return out.wholeDigits.empty()
                       ? Status::UserError("Extracted digits are empty.")
                       : Status::OK();
        }

        // Optional dot (if given in fractional form).
        if (s[pos] == '.') {
            // Extract the fractional digits.
            ++pos;
            out.fractionalDigits = detail::extractDigits(s, pos, size);
            pos += out.fractionalDigits.size();
        }

        if (out.wholeDigits.empty() && out.fractionalDigits.empty()) {
            return Status::UserError("Extracted digits are empty.");
        }
        if (pos == size) {
            return Status::OK();
        }
        // Optional exponent.
        if (s[pos] == 'e' || s[pos] == 'E') {
            ++pos;
            bool withSign = pos < size && (s[pos] == '+' || s[pos] == '-');
            if (withSign && pos == size - 1) {
                return Status::UserError("The exponent part only contains sign.");
            }
            // Make sure all chars after sign are digits, as as melon::tryTo allows
            // leading and trailing whitespaces.
            for (auto i = static_cast<size_t>(withSign); i < size - pos; ++i) {
                if (!std::isdigit(s[pos + i])) {
                    return Status::UserError(
                        "Non-digit character '{}' is not allowed in the exponent part.",
                        s[pos + i]);
                }
            }
            out.exponent = melon::to<int32_t>(melon::StringPiece(s + pos, size - pos));
            return Status::OK();
        }
        return pos == size
                   ? Status::OK()
                   : Status::UserError(
                       "Chars '{}' are invalid.", std::string(s + pos, size - pos));
    }

    Status detail::parseHugeInt(
        const DecimalComponents &decimalComponents,
        int128_t &out) {
        // Parse the whole digits.
        if (decimalComponents.wholeDigits.size() > 0) {
            const auto tryValue = melon::tryTo<int128_t>(melon::StringPiece(
                decimalComponents.wholeDigits.data(),
                decimalComponents.wholeDigits.size()));
            if (tryValue.hasError()) {
                return Status::UserError("Value too large.");
            }
            out = tryValue.value();
        }

        // Parse the fractional digits.
        if (decimalComponents.fractionalDigits.size() > 0) {
            const auto length = decimalComponents.fractionalDigits.size();
            bool overflow =
                    __builtin_mul_overflow(out, DecimalUtil::kPowersOfTen[length], &out);
            if (overflow) {
                return Status::UserError("Value too large.");
            }
            const auto tryValue = melon::tryTo<int128_t>(
                melon::StringPiece(decimalComponents.fractionalDigits.data(), length));
            if (tryValue.hasError()) {
                return Status::UserError("Value too large.");
            }
            overflow = __builtin_add_overflow(out, tryValue.value(), &out);
            POLLUX_DCHECK(!overflow);
        }
        return Status::OK();
    }

    namespace {
        const tz::TimeZone *getTimeZoneFromConfig(const core::QueryConfig &config) {
            if (config.adjustTimestampToTimezone()) {
                const auto sessionTzName = config.sessionTimezone();
                if (!sessionTzName.empty()) {
                    return tz::locateZone(sessionTzName);
                }
            }
            return nullptr;
        }
    } // namespace

    VectorPtr CastExpr::castFromDate(
        const SelectivityVector &rows,
        const BaseVector &input,
        exec::EvalCtx &context,
        const TypePtr &toType) {
        VectorPtr castResult;
        context.ensure_writable(rows, toType, castResult);
        (*castResult).clear_nulls(rows);

        auto *inputFlatVector = input.as<SimpleVector<int32_t> >();
        switch (toType->kind()) {
            case TypeKind::VARCHAR: {
                auto *resultFlatVector = castResult->as<FlatVector<StringView> >();
                applyToSelectedNoThrowLocal(context, rows, castResult, [&](int row) {
                    try {
                        // TODO Optimize to avoid creating an intermediate string.
                        auto output = DATE()->toString(inputFlatVector->value_at(row));
                        auto writer = exec::StringWriter(resultFlatVector, row);
                        writer.resize(output.size());
                        ::memcpy(writer.data(), output.data(), output.size());
                        writer.finalize();
                    } catch (const PolluxException &ue) {
                        if (!ue.isUserError()) {
                            throw;
                        }
                        POLLUX_USER_FAIL(
                            makeErrorMessage(input, row, toType) + " " + ue.message());
                    } catch (const std::exception &e) {
                        POLLUX_USER_FAIL(
                            makeErrorMessage(input, row, toType) + " " + e.what());
                    }
                });
                return castResult;
            }
            case TypeKind::TIMESTAMP: {
                static const int64_t kMillisPerDay{86'400'000};
                const auto *timeZone =
                        getTimeZoneFromConfig(context.execCtx()->queryCtx()->queryConfig());
                auto *resultFlatVector = castResult->as<FlatVector<Timestamp> >();
                applyToSelectedNoThrowLocal(context, rows, castResult, [&](int row) {
                    auto timestamp = Timestamp::fromMillis(
                        inputFlatVector->value_at(row) * kMillisPerDay);
                    if (timeZone) {
                        timestamp.toGMT(*timeZone);
                    }
                    resultFlatVector->set(row, timestamp);
                });

                return castResult;
            }
            default:
                POLLUX_UNSUPPORTED(
                    "Cast from DATE to {} is not supported", toType->toString());
        }
    }

    VectorPtr CastExpr::castToDate(
        const SelectivityVector &rows,
        const BaseVector &input,
        exec::EvalCtx &context,
        const TypePtr &fromType) {
        VectorPtr castResult;
        context.ensure_writable(rows, DATE(), castResult);
        (*castResult).clear_nulls(rows);
        auto *resultFlatVector = castResult->as<FlatVector<int32_t> >();
        switch (fromType->kind()) {
            case TypeKind::VARCHAR: {
                auto *inputVector = input.as<SimpleVector<StringView> >();
                applyToSelectedNoThrowLocal(context, rows, castResult, [&](int row) {
                    bool wrapException = true;
                    try {
                        const auto result =
                                hooks_->castStringToDate(inputVector->value_at(row));
                        if (result.hasError()) {
                            wrapException = false;
                            if (setNullInResultAtError()) {
                                resultFlatVector->set_null(row, true);
                            } else {
                                if (context.captureErrorDetails()) {
                                    context.setStatus(
                                        row,
                                        Status::UserError(
                                            "{} {}",
                                            makeErrorMessage(input, row, DATE()),
                                            result.error().message()));
                                } else {
                                    context.setStatus(row, Status::UserError());
                                }
                            }
                        } else {
                            resultFlatVector->set(row, result.value());
                        }
                    } catch (const PolluxUserError &ue) {
                        if (!wrapException) {
                            throw;
                        }
                        POLLUX_USER_FAIL(
                            makeErrorMessage(input, row, DATE()) + " " + ue.message());
                    } catch (const std::exception &e) {
                        POLLUX_USER_FAIL(
                            makeErrorMessage(input, row, DATE()) + " " + e.what());
                    }
                });

                return castResult;
            }
            case TypeKind::TIMESTAMP: {
                auto *inputVector = input.as<SimpleVector<Timestamp> >();
                const auto *timeZone =
                        getTimeZoneFromConfig(context.execCtx()->queryCtx()->queryConfig());
                applyToSelectedNoThrowLocal(context, rows, castResult, [&](int row) {
                    const auto days = util::toDate(inputVector->value_at(row), timeZone);
                    resultFlatVector->set(row, days);
                });
                return castResult;
            }
            default:
                POLLUX_UNSUPPORTED(
                    "Cast from {} to DATE is not supported", fromType->toString());
        }
    }

    VectorPtr CastExpr::castFromIntervalDayTime(
        const SelectivityVector &rows,
        const BaseVector &input,
        exec::EvalCtx &context,
        const TypePtr &toType) {
        VectorPtr castResult;
        context.ensure_writable(rows, toType, castResult);
        (*castResult).clear_nulls(rows);

        auto *inputFlatVector = input.as<SimpleVector<int64_t> >();
        switch (toType->kind()) {
            case TypeKind::VARCHAR: {
                auto *resultFlatVector = castResult->as<FlatVector<StringView> >();
                applyToSelectedNoThrowLocal(context, rows, castResult, [&](int row) {
                    try {
                        // TODO Optimize to avoid creating an intermediate string.
                        auto output =
                                INTERVAL_DAY_TIME()->valueToString(inputFlatVector->value_at(row));
                        auto writer = exec::StringWriter(resultFlatVector, row);
                        writer.resize(output.size());
                        ::memcpy(writer.data(), output.data(), output.size());
                        writer.finalize();
                    } catch (const PolluxException &ue) {
                        if (!ue.isUserError()) {
                            throw;
                        }
                        POLLUX_USER_FAIL(
                            makeErrorMessage(input, row, toType) + " " + ue.message());
                    } catch (const std::exception &e) {
                        POLLUX_USER_FAIL(
                            makeErrorMessage(input, row, toType) + " " + e.what());
                    }
                });
                return castResult;
            }
            default:
                POLLUX_UNSUPPORTED(
                    "Cast from {} to {} is not supported",
                    INTERVAL_DAY_TIME()->toString(),
                    toType->toString());
        }
    }

    namespace {
        void propagateErrorsOrSetNulls(
            bool setNullInResultAtError,
            EvalCtx &context,
            const SelectivityVector &nestedRows,
            const BufferPtr &elementToTopLevelRows,
            VectorPtr &result,
            EvalErrorsPtr &oldErrors) {
            if (context.errors()) {
                if (setNullInResultAtError) {
                    // Errors in context.errors() should be translated to nulls in the top
                    // level rows.
                    context.convertElementErrorsToTopLevelNulls(
                        nestedRows, elementToTopLevelRows, result);
                } else {
                    context.addElementErrorsToTopLevel(
                        nestedRows, elementToTopLevelRows, oldErrors);
                }
            }
        }
    } // namespace

#define POLLUX_DYNAMIC_DECIMAL_TYPE_DISPATCH(       \
    TEMPLATE_FUNC, decimalTypePtr, ...)            \
  [&]() {                                          \
    if (decimalTypePtr->isLongDecimal()) {         \
      return TEMPLATE_FUNC<int128_t>(__VA_ARGS__); \
    } else {                                       \
      return TEMPLATE_FUNC<int64_t>(__VA_ARGS__);  \
    }                                              \
  }()

    VectorPtr CastExpr::applyMap(
        const SelectivityVector &rows,
        const MapVector *input,
        exec::EvalCtx &context,
        const MapType &fromType,
        const MapType &toType) {
        // Cast input keys/values vector to output keys/values vector using their
        // element selectivity vector

        // Initialize nested rows
        auto mapKeys = input->mapKeys();
        auto mapValues = input->mapValues();

        SelectivityVector nestedRows;
        BufferPtr elementToTopLevelRows;
        if (fromType.keyType() != toType.keyType() ||
            fromType.valueType() != toType.valueType()) {
            nestedRows = functions::toElementRows(mapKeys->size(), rows, input);
            elementToTopLevelRows = functions::getElementToTopLevelRows(
                mapKeys->size(), rows, input, context.pool());
        }

        EvalErrorsPtr oldErrors;
        context.swapErrors(oldErrors);

        // Cast keys
        VectorPtr newMapKeys;
        if (fromType.keyType() == toType.keyType()) {
            newMapKeys = input->mapKeys();
        } else {
            {
                ScopedVarSetter holder(&inTopLevel, false);
                apply(
                    nestedRows,
                    mapKeys,
                    context,
                    fromType.keyType(),
                    toType.keyType(),
                    newMapKeys);
            }
        }

        // Cast values
        VectorPtr newMapValues;
        if (fromType.valueType() == toType.valueType()) {
            newMapValues = mapValues;
        } else {
            {
                ScopedVarSetter holder(&inTopLevel, false);
                apply(
                    nestedRows,
                    mapValues,
                    context,
                    fromType.valueType(),
                    toType.valueType(),
                    newMapValues);
            }
        }

        // Returned map vector should be addressable for every element, even those
        // that are not selected.
        BufferPtr sizes = input->sizes();
        if (newMapKeys->is_constant_encoding() && newMapValues->is_constant_encoding()) {
            // We extends size since that is cheap.
            newMapKeys->resize(input->mapKeys()->size());
            newMapValues->resize(input->mapValues()->size());
        } else if (
            newMapKeys->size() < input->mapKeys()->size() ||
            newMapValues->size() < input->mapValues()->size()) {
            sizes =
                    AlignedBuffer::allocate<vector_size_t>(rows.end(), context.pool(), 0);
            auto *inputSizes = input->rawSizes();
            auto *rawSizes = sizes->asMutable<vector_size_t>();

            rows.applyToSelected(
                [&](vector_size_t row) { rawSizes[row] = inputSizes[row]; });
        }

        // Assemble the output map
        VectorPtr result = std::make_shared<MapVector>(
            context.pool(),
            MAP(toType.keyType(), toType.valueType()),
            input->nulls(),
            rows.end(),
            input->offsets(),
            sizes,
            newMapKeys,
            newMapValues);

        propagateErrorsOrSetNulls(
            setNullInResultAtError(),
            context,
            nestedRows,
            elementToTopLevelRows,
            result,
            oldErrors);

        // Restore original state.
        context.swapErrors(oldErrors);
        return result;
    }

    VectorPtr CastExpr::applyArray(
        const SelectivityVector &rows,
        const ArrayVector *input,
        exec::EvalCtx &context,
        const ArrayType &fromType,
        const ArrayType &toType) {
        // Cast input array elements to output array elements based on their types
        // using their linear selectivity vector
        auto arrayElements = input->elements();

        auto nestedRows =
                functions::toElementRows(arrayElements->size(), rows, input);
        auto elementToTopLevelRows = functions::getElementToTopLevelRows(
            arrayElements->size(), rows, input, context.pool());

        EvalErrorsPtr oldErrors;
        context.swapErrors(oldErrors);

        VectorPtr newElements; {
            ScopedVarSetter holder(&inTopLevel, false);
            apply(
                nestedRows,
                arrayElements,
                context,
                fromType.elementType(),
                toType.elementType(),
                newElements);
        }

        // Returned array vector should be addressable for every element, even those
        // that are not selected.
        BufferPtr sizes = input->sizes();
        if (newElements->is_constant_encoding()) {
            // If the newElements we extends its size since that is cheap.
            newElements->resize(input->elements()->size());
        } else if (newElements->size() < input->elements()->size()) {
            sizes =
                    AlignedBuffer::allocate<vector_size_t>(rows.end(), context.pool(), 0);
            auto *inputSizes = input->rawSizes();
            auto *rawSizes = sizes->asMutable<vector_size_t>();
            rows.applyToSelected(
                [&](vector_size_t row) { rawSizes[row] = inputSizes[row]; });
        }

        VectorPtr result = std::make_shared<ArrayVector>(
            context.pool(),
            ARRAY(toType.elementType()),
            input->nulls(),
            rows.end(),
            input->offsets(),
            sizes,
            newElements);

        propagateErrorsOrSetNulls(
            setNullInResultAtError(),
            context,
            nestedRows,
            elementToTopLevelRows,
            result,
            oldErrors);
        // Restore original state.
        context.swapErrors(oldErrors);
        return result;
    }

    VectorPtr CastExpr::applyRow(
        const SelectivityVector &rows,
        const RowVector *input,
        exec::EvalCtx &context,
        const RowType &fromType,
        const TypePtr &toType) {
        const RowType &toRowType = toType->as_row();
        int numInputChildren = input->children().size();
        int numOutputChildren = toRowType.size();

        // Extract the flag indicating matching of children must be done by name or
        // position
        auto matchByName =
                context.execCtx()->queryCtx()->queryConfig().isMatchStructByName();

        // Cast each row child to its corresponding output child
        std::vector<VectorPtr> newChildren;
        newChildren.reserve(numOutputChildren);

        EvalErrorsPtr oldErrors;
        if (setNullInResultAtError()) {
            // We need to isolate errors that happen during the cast from previous
            // errors since those translate to nulls, unlike exisiting errors.
            context.swapErrors(oldErrors);
        }

        for (auto toChildrenIndex = 0; toChildrenIndex < numOutputChildren;
             toChildrenIndex++) {
            // For each child, find the corresponding column index in the output
            const auto &toFieldName = toRowType.nameOf(toChildrenIndex);
            bool matchNotFound = false;

            // If match is by field name and the input field name is not found
            // in the output row type, do not consider it in the output
            int fromChildrenIndex = -1;
            if (matchByName) {
                if (!fromType.containsChild(toFieldName)) {
                    matchNotFound = true;
                } else {
                    fromChildrenIndex = fromType.getChildIdx(toFieldName);
                    toChildrenIndex = toRowType.getChildIdx(toFieldName);
                }
            } else {
                fromChildrenIndex = toChildrenIndex;
                if (fromChildrenIndex >= numInputChildren) {
                    matchNotFound = true;
                }
            }

            // Updating output types and names
            VectorPtr outputChild;
            const auto &toChildType = toRowType.childAt(toChildrenIndex);

            if (matchNotFound) {
                // Create a vector for null for this child
                context.ensure_writable(rows, toChildType, outputChild);
                outputChild->add_nulls(rows);
            } else {
                const auto &inputChild = input->children()[fromChildrenIndex];
                if (toChildType == inputChild->type()) {
                    outputChild = inputChild;
                } else {
                    // Apply cast for the child.
                    ScopedVarSetter holder(&inTopLevel, false);
                    apply(
                        rows,
                        inputChild,
                        context,
                        inputChild->type(),
                        toChildType,
                        outputChild);
                }
            }
            newChildren.emplace_back(std::move(outputChild));
        }

        // Assemble the output row
        VectorPtr result = std::make_shared<RowVector>(
            context.pool(),
            toType,
            input->nulls(),
            rows.end(),
            std::move(newChildren));

        if (setNullInResultAtError()) {
            // Set errors as nulls.
            if (auto errors = context.errors()) {
                rows.applyToSelected([&](auto row) {
                    if (errors->hasErrorAt(row)) {
                        result->set_null(row, true);
                    }
                });
            }
            // Restore original state.
            context.swapErrors(oldErrors);
        }

        return result;
    }

    template<typename toDecimalType>
    VectorPtr CastExpr::applyDecimal(
        const SelectivityVector &rows,
        const BaseVector &input,
        exec::EvalCtx &context,
        const TypePtr &fromType,
        const TypePtr &toType) {
        VectorPtr castResult;
        context.ensure_writable(rows, toType, castResult);
        (*castResult).clear_nulls(rows);

        // toType is a decimal
        switch (fromType->kind()) {
            case TypeKind::BOOLEAN:
                applyIntToDecimalCastKernel<bool, toDecimalType>(
                    rows, input, context, toType, castResult);
                break;
            case TypeKind::TINYINT:
                applyIntToDecimalCastKernel<int8_t, toDecimalType>(
                    rows, input, context, toType, castResult);
                break;
            case TypeKind::SMALLINT:
                applyIntToDecimalCastKernel<int16_t, toDecimalType>(
                    rows, input, context, toType, castResult);
                break;
            case TypeKind::INTEGER:
                applyIntToDecimalCastKernel<int32_t, toDecimalType>(
                    rows, input, context, toType, castResult);
                break;
            case TypeKind::REAL:
                applyFloatingPointToDecimalCastKernel<float, toDecimalType>(
                    rows, input, context, toType, castResult);
                break;
            case TypeKind::DOUBLE:
                applyFloatingPointToDecimalCastKernel<double, toDecimalType>(
                    rows, input, context, toType, castResult);
                break;
            case TypeKind::BIGINT: {
                if (fromType->isShortDecimal()) {
                    applyDecimalCastKernel<int64_t, toDecimalType>(
                        rows, input, context, fromType, toType, castResult);
                    break;
                }
                applyIntToDecimalCastKernel<int64_t, toDecimalType>(
                    rows, input, context, toType, castResult);
                break;
            }
            case TypeKind::HUGEINT: {
                if (fromType->isLongDecimal()) {
                    applyDecimalCastKernel<int128_t, toDecimalType>(
                        rows, input, context, fromType, toType, castResult);
                    break;
                }
                [[fallthrough]];
            }
            case TypeKind::VARCHAR:
                applyVarcharToDecimalCastKernel<toDecimalType>(
                    rows, input, context, toType, castResult);
                break;
            default:
                POLLUX_UNSUPPORTED(
                    "Cast from {} to {} is not supported",
                    fromType->toString(),
                    toType->toString());
        }
        return castResult;
    }

    void CastExpr::applyPeeled(
        const SelectivityVector &rows,
        const BaseVector &input,
        exec::EvalCtx &context,
        const TypePtr &fromType,
        const TypePtr &toType,
        VectorPtr &result) {
        auto castFromOperator = getCastOperator(fromType);
        if (castFromOperator && !castFromOperator->isSupportedToType(toType)) {
            POLLUX_USER_FAIL(
                "Cannot cast {} to {}.", fromType->toString(), toType->toString());
        }

        auto castToOperator = getCastOperator(toType);
        if (castToOperator && !castToOperator->isSupportedFromType(fromType)) {
            POLLUX_USER_FAIL(
                "Cannot cast {} to {}.", fromType->toString(), toType->toString());
        }

        if (castFromOperator || castToOperator) {
            POLLUX_USER_CHECK(
                *fromType != *toType,
                "Attempting to cast from {} to itself.",
                fromType->toString());

            auto applyCustomCast = [&]() {
                if (castToOperator) {
                    castToOperator->castTo(input, context, rows, toType, result, hooks_);
                } else {
                    castFromOperator->castFrom(input, context, rows, toType, result);
                }
            };

            if (setNullInResultAtError()) {
                // This can be optimized by passing setNullInResultAtError() to castTo and
                // castFrom operations.

                EvalErrorsPtr oldErrors;
                context.swapErrors(oldErrors);

                applyCustomCast();

                if (context.errors()) {
                    auto errors = context.errors();
                    auto rawNulls = result->mutable_raw_nulls();

                    rows.applyToSelected([&](auto row) {
                        if (errors->hasErrorAt(row)) {
                            bits::set_null(rawNulls, row, true);
                        }
                    });
                };
                // Restore original state.
                context.swapErrors(oldErrors);
            } else {
                applyCustomCast();
            }
        } else if (fromType->isDate()) {
            result = castFromDate(rows, input, context, toType);
        } else if (toType->isDate()) {
            result = castToDate(rows, input, context, fromType);
        } else if (fromType->isIntervalDayTime()) {
            result = castFromIntervalDayTime(rows, input, context, toType);
        } else if (toType->isIntervalDayTime()) {
            POLLUX_UNSUPPORTED(
                "Cast from {} to {} is not supported",
                fromType->toString(),
                toType->toString());
        } else if (toType->isShortDecimal()) {
            result = applyDecimal<int64_t>(rows, input, context, fromType, toType);
        } else if (toType->isLongDecimal()) {
            result = applyDecimal<int128_t>(rows, input, context, fromType, toType);
        } else if (fromType->isDecimal()) {
            switch (toType->kind()) {
                case TypeKind::VARCHAR:
                    result = POLLUX_DYNAMIC_DECIMAL_TYPE_DISPATCH(
                                applyDecimalToVarcharCast,
                                fromType,
                                rows,
                                input,
                                context,
                                fromType);
                    break;
                default:
                    result = POLLUX_DYNAMIC_DECIMAL_TYPE_DISPATCH(
                                applyDecimalToPrimitiveCast,
                                fromType,
                                rows,
                                input,
                                context,
                                fromType,
                                toType);
            }
        } else if (
            fromType->kind() == TypeKind::TIMESTAMP &&
            (toType->kind() == TypeKind::VARCHAR ||
             toType->kind() == TypeKind::VARBINARY)) {
            result = applyTimestampToVarcharCast(toType, rows, context, input);
        } else if (toType->kind() == TypeKind::VARBINARY) {
            switch (fromType->kind()) {
                case TypeKind::TINYINT:
                    result = applyIntToBinaryCast<int8_t>(rows, context, input);
                    break;
                case TypeKind::SMALLINT:
                    result = applyIntToBinaryCast<int16_t>(rows, context, input);
                    break;
                case TypeKind::INTEGER:
                    result = applyIntToBinaryCast<int32_t>(rows, context, input);
                    break;
                case TypeKind::BIGINT:
                    result = applyIntToBinaryCast<int64_t>(rows, context, input);
                    break;
                default:
                    // Handle primitive type conversions.
                    applyCastPrimitivesDispatch<TypeKind::VARBINARY>(
                        fromType, toType, rows, context, input, result);
                    break;
            }
        } else {
            switch (toType->kind()) {
                case TypeKind::MAP:
                    result = applyMap(
                        rows,
                        input.as_unchecked<MapVector>(),
                        context,
                        fromType->as_map(),
                        toType->as_map());
                    break;
                case TypeKind::ARRAY:
                    result = applyArray(
                        rows,
                        input.as_unchecked<ArrayVector>(),
                        context,
                        fromType->as_array(),
                        toType->as_array());
                    break;
                case TypeKind::ROW:
                    result = applyRow(
                        rows,
                        input.as_unchecked<RowVector>(),
                        context,
                        fromType->as_row(),
                        toType);
                    break;
                default: {
                    // Handle primitive type conversions.
                    POLLUX_DYNAMIC_SCALAR_TYPE_DISPATCH(
                        applyCastPrimitivesDispatch,
                        toType->kind(),
                        fromType,
                        toType,
                        rows,
                        context,
                        input,
                        result);
                }
            }
        }
    }

    VectorPtr CastExpr::applyTimestampToVarcharCast(
        const TypePtr &toType,
        const SelectivityVector &rows,
        exec::EvalCtx &context,
        const BaseVector &input) {
        VectorPtr result;
        context.ensure_writable(rows, toType, result);
        (*result).clear_nulls(rows);
        auto flatResult = result->as_flat_vector<StringView>();
        const auto simpleInput = input.as<SimpleVector<Timestamp> >();

        const auto &options = hooks_->timestampToStringOptions();
        const uint32_t rowSize = getMaxStringLength(options);

        Buffer *buffer = flatResult->getBufferWithSpace(
            rows.countSelected() * rowSize, true /*exactSize*/);
        char *rawBuffer = buffer->asMutable<char>() + buffer->size();

        applyToSelectedNoThrowLocal(context, rows, result, [&](vector_size_t row) {
            // Adjust input timestamp according the session timezone.
            Timestamp inputValue(simpleInput->value_at(row));
            if (options.timeZone) {
                inputValue.toTimezone(*(options.timeZone));
            }
            const auto stringView =
                    Timestamp::tsToStringView(inputValue, options, rawBuffer);
            flatResult->setNoCopy(row, stringView);
            // The result of both Presto and Spark contains more than 12 digits even
            // when 'zeroPaddingYear' is disabled.
            POLLUX_DCHECK(!stringView.isInline());
            rawBuffer += stringView.size();
        });

        // Update the exact buffer size.
        buffer->setSize(rawBuffer - buffer->asMutable<char>());
        return result;
    }

    template<typename TInput>
    VectorPtr CastExpr::applyIntToBinaryCast(
        const SelectivityVector &rows,
        exec::EvalCtx &context,
        const BaseVector &input) {
        auto result = BaseVector::create(VARBINARY(), rows.end(), context.pool());
        const auto flatResult = result->as_flat_vector<StringView>();
        const auto simpleInput = input.as<SimpleVector<TInput> >();

        // The created string view is always inlined for int types.
        char inlined[sizeof(TInput)];
        applyToSelectedNoThrowLocal(context, rows, result, [&](vector_size_t row) {
            TInput input = simpleInput->value_at(row);
            if constexpr (std::is_same_v<TInput, int8_t>) {
                inlined[0] = static_cast<char>(input & 0xFF);
            } else {
                for (int i = sizeof(TInput) - 1; i >= 0; --i) {
                    inlined[i] = static_cast<char>(input & 0xFF);
                    input >>= 8;
                }
            }
            const auto stringView = StringView(inlined, sizeof(TInput));
            flatResult->setNoCopy(row, stringView);
        });

        return result;
    }

    void CastExpr::apply(
        const SelectivityVector &rows,
        const VectorPtr &input,
        exec::EvalCtx &context,
        const TypePtr &fromType,
        const TypePtr &toType,
        VectorPtr &result) {
        LocalSelectivityVector remainingRows(context, rows);

        context.deselectErrors(*remainingRows);

        LocalDecodedVector decoded(context, *input, *remainingRows);
        auto *rawNulls = decoded->nulls(remainingRows.get());

        if (rawNulls) {
            remainingRows->deselectNulls(
                rawNulls, remainingRows->begin(), remainingRows->end());
        }

        VectorPtr localResult;
        if (!remainingRows->hasSelections()) {
            localResult =
                    BaseVector::create_null_constant(toType, rows.end(), context.pool());
        } else if (decoded->isIdentityMapping()) {
            applyPeeled(
                *remainingRows,
                *decoded->base(),
                context,
                fromType,
                toType,
                localResult);
        } else {
            withContextSaver([&](ContextSaver &saver) {
                LocalSelectivityVector newRowsHolder(*context.execCtx());

                LocalDecodedVector localDecoded(context);
                std::vector<VectorPtr> peeledVectors;
                auto peeledEncoding = PeeledEncoding::peel(
                    {input}, *remainingRows, localDecoded, true, peeledVectors);
                POLLUX_CHECK_EQ(peeledVectors.size(), 1);
                if (peeledVectors[0]->is_lazy()) {
                    peeledVectors[0] =
                            peeledVectors[0]->as<LazyVector>()->loaded_vector_shared();
                }
                auto newRows =
                        peeledEncoding->translateToInnerRows(*remainingRows, newRowsHolder);
                // Save context and set the peel.
                context.saveAndReset(saver, *remainingRows);
                context.setPeeledEncoding(peeledEncoding);
                applyPeeled(
                    *newRows, *peeledVectors[0], context, fromType, toType, localResult);

                localResult = context.getPeeledEncoding()->wrap(
                    toType, context.pool(), localResult, *remainingRows);
            });
        }
        context.moveOrCopyResult(localResult, *remainingRows, result);
        context.releaseVector(localResult);

        // If there are nulls or rows that encountered errors in the input, add nulls
        // to the result at the same rows.
        POLLUX_CHECK_NOT_NULL(result);
        if (rawNulls || context.errors()) {
            EvalCtx::add_nulls(
                rows, remainingRows->asRange().bits(), context, toType, result);
        }
    }

    void CastExpr::evalSpecialForm(
        const SelectivityVector &rows,
        EvalCtx &context,
        VectorPtr &result) {
        VectorPtr input;
        inputs_[0]->eval(rows, context, input);
        auto fromType = inputs_[0]->type();
        auto toType = std::const_pointer_cast<const Type>(type_);

        inTopLevel = true;
        if (nullOnFailure()) {
            ScopedVarSetter holder{context.mutableThrowOnError(), false};
            ScopedVarSetter captureErrorDetails(
                context.mutableCaptureErrorDetails(), false);

            ScopedThreadSkipErrorDetails skipErrorDetails(true);

            apply(rows, input, context, fromType, toType, result);
        } else {
            apply(rows, input, context, fromType, toType, result);
        }
        // Return 'input' back to the vector pool in 'context' so it can be reused.
        context.releaseVector(input);
    }

    std::string CastExpr::toString(bool recursive) const {
        std::stringstream out;
        out << name() << "(";
        if (recursive) {
            appendInputs(out);
        } else {
            out << inputs_[0]->toString(false);
        }
        out << " as " << type_->toString() << ")";
        return out.str();
    }

    std::string CastExpr::toSql(std::vector<VectorPtr> *complexConstants) const {
        std::stringstream out;
        out << name() << "(";
        appendInputsSql(out, complexConstants);
        out << " as ";
        toTypeSql(type_, out);
        out << ")";
        return out.str();
    }

    CastOperatorPtr CastExpr::getCastOperator(const TypePtr &type) {
        const auto *key = type->name();

        auto it = castOperators_.find(key);
        if (it != castOperators_.end()) {
            return it->second;
        }

        auto castOperator = getCustomTypeCastOperator(key);
        if (castOperator == nullptr) {
            return nullptr;
        }

        castOperators_.emplace(key, castOperator);
        return castOperator;
    }

    TypePtr CastCallToSpecialForm::resolveType(
        const std::vector<TypePtr> & /* argTypes */) {
        POLLUX_FAIL("CAST expressions do not support type resolution.");
    }

    ExprPtr CastCallToSpecialForm::constructSpecialForm(
        const TypePtr &type,
        std::vector<ExprPtr> &&compiledChildren,
        bool trackCpuUsage,
        const core::QueryConfig &config) {
        POLLUX_CHECK_EQ(
            compiledChildren.size(),
            1,
            "CAST statements expect exactly 1 argument, received {}.",
            compiledChildren.size());
        const auto inputKind = compiledChildren[0]->type()->kind();
        if (type->kind() == TypeKind::VARBINARY &&
            (inputKind == TypeKind::TINYINT || inputKind == TypeKind::SMALLINT ||
             inputKind == TypeKind::INTEGER || inputKind == TypeKind::BIGINT)) {
            POLLUX_UNSUPPORTED(
                "Cannot cast {} to VARBINARY.",
                compiledChildren[0]->type()->toString());
        }
        return std::make_shared<CastExpr>(
            type,
            std::move(compiledChildren[0]),
            trackCpuUsage,
            false,
            std::make_shared<PrestoCastHooks>(config));
    }

    TypePtr TryCastCallToSpecialForm::resolveType(
        const std::vector<TypePtr> & /* argTypes */) {
        POLLUX_FAIL("TRY CAST expressions do not support type resolution.");
    }

    ExprPtr TryCastCallToSpecialForm::constructSpecialForm(
        const TypePtr &type,
        std::vector<ExprPtr> &&compiledChildren,
        bool trackCpuUsage,
        const core::QueryConfig &config) {
        POLLUX_CHECK_EQ(
            compiledChildren.size(),
            1,
            "TRY CAST statements expect exactly 1 argument, received {}.",
            compiledChildren.size());
        return std::make_shared<CastExpr>(
            type,
            std::move(compiledChildren[0]),
            trackCpuUsage,
            true,
            std::make_shared<PrestoCastHooks>(config));
    }
} // namespace kumo::pollux::exec
